package brokerFailureDetector;

import java.util.UUID;

import javax.jms.DeliveryMode;
import javax.jms.TopicSubscriber;

import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQSession;
import org.apache.qpid.client.AMQTopic;
import org.apache.qpid.client.TopicPublisherAdapter;
import org.apache.qpid.client.message.AbstractJMSMessage;
import org.apache.qpid.client.message.JMSTextMessage;

/**
 * Sends AMQP messages to itself through a broker and gets an idea of how well
 * that broker is working based on whether it gets the message or not. 
 */
public class DetectorThread extends Thread {
	
	//detector
	Detector bfDetector;
	
	//unique id of the client
    protected final String CLIENT_ID;
    //name of the used topic
    protected final String TOPIC;
	//machine in which the broker to test is running
    protected final String TEST_BROKER;
    //port in which the broker to test is running
    protected final int PORT;
    
    //time in which a message must arrive, in milliseconds
    protected final int MAX_WAIT = 5000;
    //time between messages
    protected final int INTERVAL = 5000;
    
    /**
     * Constructor.
     * 
     * @param bfDetector  to report the state of the broker to
     * @param test_broker  IP address of the broker to test
     * @param port  port in which the broker to test is listening
     */
    public DetectorThread(Detector bfDetector, String test_broker, int port) {
    	
    	this.bfDetector = bfDetector;
    	CLIENT_ID = UUID.randomUUID().toString();
    	TOPIC = "brokerFailure." + CLIENT_ID;
    	TEST_BROKER = test_broker;
    	PORT = port;
    	
    	start();
    	
    }
    
    /**
     * Sends a message to the broker to test to a topic it is subscribed to and
     * waits for receiving the message.
     * If it arrives in MAX_WAIT milliseconds, the state of the broker is 10.
     * If it doesn't, the state of the broker is 0.
     * In any case, it tells the detector the state of the broker.
     * The process is repeated every INTERVAL milliseconds.
     */
    public void run() {
    	while (true) {
			try {

				//Create and start a connection
				String con = "amqp://" + bfDetector.getUSERNAME() + ":"
							 + bfDetector.getPASSWORD() + "@" + CLIENT_ID + "/"
							 + bfDetector.getVIRTUAL_HOST()
							 + "?brokerlist='tcp://" + TEST_BROKER + ":"
							 + PORT + "'";
				AMQConnection connection = new AMQConnection(con);
				connection.start();
				//Create a session, non transacted and auto acknowledged
				AMQSession session = (AMQSession) connection.createSession(
						false, AMQSession.AUTO_ACKNOWLEDGE);

				//Create topicURL
				String tURL = "topic://" + bfDetector.getEXCHANGE()
							  + "/?routingkey='" + TOPIC
							  + "'&exclusive='true'&autodelete='true'";
				//Create topic
				AMQTopic topic = (AMQTopic) session.createTopic(tURL);

				//Get a random subscription id
				String subscriptionId = UUID.randomUUID().toString();
				//Create a subscriber
				TopicSubscriber subscriber = session.createDurableSubscriber(
											 topic, subscriptionId);
//				//Log
//				System.out.println("Thread in " + controller.getHOST_BROKER()
//						  + " testing " + TEST_BROKER + " subscribed to topic "
//						  + subscriber.getTopic());

				//Create publisher
				TopicPublisherAdapter publisher = (TopicPublisherAdapter)
												session.createPublisher(topic);
				//Create the empty message to send
				JMSTextMessage txtMessage = (JMSTextMessage) session
						.createTextMessage("");
				while (true) {
					//Send the message created
					publisher.publish(txtMessage, DeliveryMode.PERSISTENT,
							AbstractJMSMessage.DEFAULT_PRIORITY,
							AbstractJMSMessage.DEFAULT_TIME_TO_LIVE);
//					//Log
//					System.out.println("Thread in "
//					  + bfDetector.getHOST_BROKER() + " testing " + TEST_BROKER
//					  + " sent message to " + txtMessage.getJMSDestination());
					//Receive a message if it arrives in the specified time
					JMSTextMessage recMsg = (JMSTextMessage) subscriber
							.receive(MAX_WAIT);
					if (recMsg == null) {
						throw new Exception(
								"No message received in the specified time");
					} else {
//						System.out.println("Thread in "
//									+ controller.getHOST_BROKER() + " testing "
//									+ TEST_BROKER + " received message \""
//									+ recMsg.getText() + "\" from "
//									+ recMsg.getJMSDestination());
						//TODO: different numbers depending on the delay
//						System.out.println("Thread in "
//									+ bfDetector.getHOST_BROKER() + " testing "
//									+ TEST_BROKER + ": OK");
						bfDetector.updateState(TEST_BROKER, 10);
					}
					//Sleep a bit
					sleep(INTERVAL);
				}

			} catch (Exception e) {
//				e.printStackTrace();
//				System.err.println("Thread in " + bfDetector.getHOST_BROKER()
//								   + " testing " + TEST_BROKER + ": FAIL");
				bfDetector.updateState(TEST_BROKER, 0);
				try {
					sleep(INTERVAL);
				} catch (InterruptedException e1) {
//					e1.printStackTrace();
				}
			}
		}
    }

}
